package com.atguigu.edu.realtime.utils;

import com.atguigu.edu.realtime.beans.TableProcess;
import com.atguigu.edu.realtime.beans.TransientSink;
import com.atguigu.edu.realtime.common.EduConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class ClickHouseUtil {
    public static <T> SinkFunction<T> getSinkFunction(String sql){
            return JdbcSink.<T>sink(sql, new JdbcStatementBuilder<T>() {
                        @Override
                        public void accept(PreparedStatement ps, T obj) throws SQLException {
                            int skipNum = 0;
                            //使用反射得到类中的所有属性(getDeclaredFields 包括私有属性)
                            Field[] declaredFields = obj.getClass().getDeclaredFields();
                            for (int i = 0; i < declaredFields.length; i++) {
                                Field declaredField = declaredFields[i];

                                TableProcess.TransientSink transientSink = declaredField.getAnnotation(TableProcess.TransientSink.class);
                                if (transientSink != null) {
                                    skipNum++;
                                    continue;
                                }
                                //设置所有私有属性公有化 暴力破解
                                declaredField.setAccessible(true);
                                try {
                                    Object value = declaredField.get(obj);
                                    ps.setObject(i + 1 - skipNum, value);
                                } catch (IllegalAccessException e) {
                                    throw new RuntimeException(e);
                                }
                            }

                        }
                    },new JdbcExecutionOptions.Builder()
                            .withBatchSize(5)
                            .withBatchIntervalMs(3000)
                            .build()
                    ,new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                            .withDriverName(EduConfig.CLICKHOUSE_DRIVER)
                            .withUrl(EduConfig.CLICKHOUSE_URL)
                            .build()
            );

        }
    }


